/*-------------------------------------------------------------------------
 *
 * parallel.c
 *	  启动并行工作程序的基础设施
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/access/transam/parallel.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/nbtree.h"
#include "access/parallel.h"
#include "access/session.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/index.h"
#include "catalog/namespace.h"
#include "catalog/pg_enum.h"
#include "catalog/storage.h"
#include "commands/async.h"
#include "commands/vacuum.h"
#include "executor/execParallel.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/sinval.h"
#include "storage/spin.h"
#include "tcop/tcopprot.h"
#include "utils/combocid.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/relmapper.h"
#include "utils/snapmgr.h"
#include "utils/typcache.h"

/*
 * 我们不想在一个错误队列上浪费大量内存，因为大多数时候，它只会处理少量小消息。然而，最好能让它足够大，以便可以在不阻塞的情况下发送典型的 ErrorResponse。这样，出现错误的工作线程可以将整个消息写入队列并终止，而无需等待用户后端。
 */
#define PARALLEL_ERROR_QUEUE_SIZE			16384

/* 并行上下文的魔术数字 TOC。 */
#define PARALLEL_MAGIC						0x50477c7c

/*
 * 每个上下文的并行状态共享的魔术数字。更高层的代码应使用更小的值，将这些非常大的值留给该模块使用。
 */
#define PARALLEL_KEY_FIXED					UINT64CONST(0xFFFFFFFFFFFF0001)
#define PARALLEL_KEY_ERROR_QUEUE			UINT64CONST(0xFFFFFFFFFFFF0002)
#define PARALLEL_KEY_LIBRARY				UINT64CONST(0xFFFFFFFFFFFF0003)
#define PARALLEL_KEY_GUC					UINT64CONST(0xFFFFFFFFFFFF0004)
#define PARALLEL_KEY_COMBO_CID				UINT64CONST(0xFFFFFFFFFFFF0005)
#define PARALLEL_KEY_TRANSACTION_SNAPSHOT	UINT64CONST(0xFFFFFFFFFFFF0006)
#define PARALLEL_KEY_ACTIVE_SNAPSHOT		UINT64CONST(0xFFFFFFFFFFFF0007)
#define PARALLEL_KEY_TRANSACTION_STATE		UINT64CONST(0xFFFFFFFFFFFF0008)
#define PARALLEL_KEY_ENTRYPOINT				UINT64CONST(0xFFFFFFFFFFFF0009)
#define PARALLEL_KEY_SESSION_DSM			UINT64CONST(0xFFFFFFFFFFFF000A)
#define PARALLEL_KEY_PENDING_SYNCS			UINT64CONST(0xFFFFFFFFFFFF000B)
#define PARALLEL_KEY_REINDEX_STATE			UINT64CONST(0xFFFFFFFFFFFF000C)
#define PARALLEL_KEY_RELMAPPER_STATE		UINT64CONST(0xFFFFFFFFFFFF000D)
#define PARALLEL_KEY_UNCOMMITTEDENUMS		UINT64CONST(0xFFFFFFFFFFFF000E)

/* 固定大小的并行状态。 */
typedef struct FixedParallelState
{
	/* 工作线程必须恢复的固定大小状态。 */
	Oid			database_id;
	Oid			authenticated_user_id;
	Oid			session_user_id;
	Oid			outer_user_id;
	Oid			current_user_id;
	Oid			temp_namespace_id;
	Oid			temp_toast_namespace_id;
	int			sec_context;
	bool		authenticated_user_is_superuser;
	bool		session_user_is_superuser;
	bool		role_is_superuser;
	PGPROC	   *parallel_leader_pgproc;
	pid_t		parallel_leader_pid;
	BackendId	parallel_leader_backend_id;
	TimestampTz xact_ts;
	TimestampTz stmt_ts;
	SerializableXactHandle serializable_xact_handle;

	/* 互斥锁保护剩余字段。 */
	slock_t		mutex;

	/* 任何工作线程的最大 XactLastRecEnd。 */
	XLogRecPtr	last_xlog_end;
} FixedParallelState;

/*
 * 我们的并行工作线程编号。我们将其初始化为 -1，表示我们不是并行工作线程。在并行工作线程中，它将在调用任何用户代码之前被设置为大于等于 0 且小于工作线程数量的值；每个并行工作线程将获得不同的并行工作线程编号。
 */
int			ParallelWorkerNumber = -1;

/* 是否有待接收的并行消息？ */
volatile bool ParallelMessagePending = false;

/* 我们是在初始化一个并行工作线程吗？ */
bool		InitializingParallelWorker = false;

/* 指向我们的固定并行状态的指针。 */
static FixedParallelState *MyFixedParallelState;

/* 活动并行上下文的列表。 */
static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list);

/* 来自 FixedParallelState 的后端本地数据副本。 */
static pid_t ParallelLeaderPid;

/*
 * 内部并行工作线程入口点的列表。我们需要这个，因为在下面的 LookupParallelWorkerFunction() 中解释的原因。
 */
static const struct
{
	const char *fn_name;
	parallel_worker_main_type fn_addr;
}			InternalParallelWorkers[] =

{
	{
		"ParallelQueryMain", ParallelQueryMain
	},
	{
		"_bt_parallel_build_main", _bt_parallel_build_main
	},
	{
		"parallel_vacuum_main", parallel_vacuum_main
	}
};

/* 私有函数。 */
static void fc_HandleParallelMessage(ParallelContext *fc_pcxt, int fc_i, StringInfo fc_msg);
static void fc_WaitForParallelWorkersToExit(ParallelContext *fc_pcxt);
static parallel_worker_main_type fc_LookupParallelWorkerFunction(const char *fc_libraryname, const char *fc_funcname);
static void fc_ParallelWorkerShutdown(int fc_code, Datum fc_arg);


/*
 * 建立一个新的并行上下文。这应该在进入并行模式后进行，且（除非有错误）该上下文应该在退出当前子事务之前被销毁。
 */
ParallelContext *
CreateParallelContext(const char *fc_library_name, const char *fc_function_name,
					  int fc_nworkers)
{
	MemoryContext fc_oldcontext;
	ParallelContext *fc_pcxt;

	/* 如果不在并行模式下创建并行上下文是不安全的。 */
	Assert(IsInParallelMode());

	/* 工作线程的数量应该是非负的。 */
	Assert(fc_nworkers >= 0);

	/* 我们可能在一个短暂的内存上下文中运行。 */
	fc_oldcontext = MemoryContextSwitchTo(TopTransactionContext);

	/* 初始化一个新的 ParallelContext。 */
	fc_pcxt = palloc0(sizeof(ParallelContext));
	fc_pcxt->subid = GetCurrentSubTransactionId();
	fc_pcxt->nworkers = fc_nworkers;
	fc_pcxt->nworkers_to_launch = fc_nworkers;
	fc_pcxt->library_name = pstrdup(fc_library_name);
	fc_pcxt->function_name = pstrdup(fc_function_name);
	fc_pcxt->error_context_stack = error_context_stack;
	shm_toc_initialize_estimator(&fc_pcxt->estimator);
	dlist_push_head(&pcxt_list, &fc_pcxt->node);

	/* 恢复先前的内存上下文。 */
	MemoryContextSwitchTo(fc_oldcontext);

	return fc_pcxt;
}

/*
 * 为并行上下文建立动态共享内存段，并将并行工作线程所需的状态和其他记账信息复制到其中。
 */
void InitializeParallelDSM(ParallelContext *fc_pcxt)
{
	MemoryContext fc_oldcontext;
	Size		fc_library_len = 0;
	Size		fc_guc_len = 0;
	Size		fc_combocidlen = 0;
	Size		fc_tsnaplen = 0;
	Size		fc_asnaplen = 0;
	Size		fc_tstatelen = 0;
	Size		fc_pendingsyncslen = 0;
	Size		fc_reindexlen = 0;
	Size		fc_relmapperlen = 0;
	Size		fc_uncommittedenumslen = 0;
	Size		fc_segsize = 0;
	int			fc_i;
	FixedParallelState *fc_fps;
	dsm_handle	fc_session_dsm_handle = DSM_HANDLE_INVALID;
	Snapshot	fc_transaction_snapshot = GetTransactionSnapshot();
	Snapshot	fc_active_snapshot = GetActiveSnapshot();

	/* 我们可能在一个非常短暂的内存上下文中运行。 */
	fc_oldcontext = MemoryContextSwitchTo(TopTransactionContext);

	/* 允许存储固定大小并行状态的空间。 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator, sizeof(FixedParallelState));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/*
	 * 如果我们在不可中断的情况下到达这里，启动任何工作线程都是不安全的：我们将无法处理它们发送的中断。我们可以通过假装没有请求任何工作线程来处理这个边缘情况。
	 */
	if (!INTERRUPTS_CAN_BE_PROCESSED())
		fc_pcxt->nworkers = 0;

	/*
	 * 通常，用户将请求至少一个工作进程，但如果碰巧没有请求，我们可以在这里跳过一系列操作。
	 */
	if (fc_pcxt->nworkers > 0)
	{
		/* 获取（或创建）每个会话 DSM 段的句柄。 */
		fc_session_dsm_handle = GetSessionDsmHandle();

		/*
		 * 如果我们未能创建每个会话的DSM段，那么我们可以继续，但我们不能安全地启动任何工作进程，因为它们的记录类型修饰符不兼容，因此它们无法交换元组。
		 */
		if (fc_session_dsm_handle == DSM_HANDLE_INVALID)
			fc_pcxt->nworkers = 0;
	}

	if (fc_pcxt->nworkers > 0)
	{
		/* 估算各种状态共享所需的空间。 */
		fc_library_len = EstimateLibraryStateSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_library_len);
		fc_guc_len = EstimateGUCStateSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_guc_len);
		fc_combocidlen = EstimateComboCIDStateSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_combocidlen);
		if (IsolationUsesXactSnapshot())
		{
			fc_tsnaplen = EstimateSnapshotSpace(fc_transaction_snapshot);
			shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_tsnaplen);
		}
		fc_asnaplen = EstimateSnapshotSpace(fc_active_snapshot);
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_asnaplen);
		fc_tstatelen = EstimateTransactionStateSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_tstatelen);
		shm_toc_estimate_chunk(&fc_pcxt->estimator, sizeof(dsm_handle));
		fc_pendingsyncslen = EstimatePendingSyncsSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_pendingsyncslen);
		fc_reindexlen = EstimateReindexStateSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_reindexlen);
		fc_relmapperlen = EstimateRelationMapSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_relmapperlen);
		fc_uncommittedenumslen = EstimateUncommittedEnumsSpace();
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_uncommittedenumslen);
		/* 如果在这里添加更多块，您可能需要添加键。 */
		shm_toc_estimate_keys(&fc_pcxt->estimator, 11);

		/* 估算错误队列所需的空间。 */
		StaticAssertStmt(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) ==
						 PARALLEL_ERROR_QUEUE_SIZE,
						 "parallel error queue size not buffer-aligned");
		shm_toc_estimate_chunk(&fc_pcxt->estimator,
							   mul_size(PARALLEL_ERROR_QUEUE_SIZE,
										fc_pcxt->nworkers));
		shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

		/* 估算入口点信息所需的空间。 */
		shm_toc_estimate_chunk(&fc_pcxt->estimator, strlen(fc_pcxt->library_name) +
							   strlen(fc_pcxt->function_name) + 2);
		shm_toc_estimate_keys(&fc_pcxt->estimator, 1);
	}

	/*
	 * 创建DSM并用新目录初始化。但如果用户没有请求任何工作进程，那么就不需要创建动态共享内存段；而是使用后端私有内存。
	 *
	 * 此外，如果我们无法创建动态共享内存段，因为已创建的段数达到了最大值，则退回到后端私有内存，并计划不使用任何工作进程。我们希望这种情况不会很常见，但放弃使用并行性总比彻底失败要好。
	 */
	fc_segsize = shm_toc_estimate(&fc_pcxt->estimator);
	if (fc_pcxt->nworkers > 0)
		fc_pcxt->seg = dsm_create(fc_segsize, DSM_CREATE_NULL_IF_MAXSEGMENTS);
	if (fc_pcxt->seg != NULL)
		fc_pcxt->toc = shm_toc_create(PARALLEL_MAGIC,
								   dsm_segment_address(fc_pcxt->seg),
								   fc_segsize);
	else
	{
		fc_pcxt->nworkers = 0;
		fc_pcxt->private_memory = MemoryContextAlloc(TopMemoryContext, fc_segsize);
		fc_pcxt->toc = shm_toc_create(PARALLEL_MAGIC, fc_pcxt->private_memory,
								   fc_segsize);
	}

	/* 在共享内存中初始化固定大小的状态。 */
	fc_fps = (FixedParallelState *)
		shm_toc_allocate(fc_pcxt->toc, sizeof(FixedParallelState));
	fc_fps->database_id = MyDatabaseId;
	fc_fps->authenticated_user_id = GetAuthenticatedUserId();
	fc_fps->session_user_id = GetSessionUserId();
	fc_fps->outer_user_id = GetCurrentRoleId();
	GetUserIdAndSecContext(&fc_fps->current_user_id, &fc_fps->sec_context);
	fc_fps->authenticated_user_is_superuser = GetAuthenticatedUserIsSuperuser();
	fc_fps->session_user_is_superuser = GetSessionUserIsSuperuser();
	fc_fps->role_is_superuser = session_auth_is_superuser;
	GetTempNamespaceState(&fc_fps->temp_namespace_id,
						  &fc_fps->temp_toast_namespace_id);
	fc_fps->parallel_leader_pgproc = MyProc;
	fc_fps->parallel_leader_pid = MyProcPid;
	fc_fps->parallel_leader_backend_id = MyBackendId;
	fc_fps->xact_ts = GetCurrentTransactionStartTimestamp();
	fc_fps->stmt_ts = GetCurrentStatementStartTimestamp();
	fc_fps->serializable_xact_handle = ShareSerializableXact();
	SpinLockInit(&fc_fps->mutex);
	fc_fps->last_xlog_end = 0;
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_FIXED, fc_fps);

	/* 如果我们没有为任何工作进程分配预算，可以跳过其余部分。 */
	if (fc_pcxt->nworkers > 0)
	{
		char	   *fc_libraryspace;
		char	   *fc_gucspace;
		char	   *fc_combocidspace;
		char	   *fc_tsnapspace;
		char	   *fc_asnapspace;
		char	   *fc_tstatespace;
		char	   *fc_pendingsyncsspace;
		char	   *fc_reindexspace;
		char	   *fc_relmapperspace;
		char	   *fc_error_queue_space;
		char	   *fc_session_dsm_handle_space;
		char	   *fc_entrypointstate;
		char	   *fc_uncommittedenumsspace;
		Size		fc_lnamelen;

		/* 序列化我们已加载的共享库。 */
		fc_libraryspace = shm_toc_allocate(fc_pcxt->toc, fc_library_len);
		SerializeLibraryState(fc_library_len, fc_libraryspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_LIBRARY, fc_libraryspace);

		/* 序列化GUC设置。 */
		fc_gucspace = shm_toc_allocate(fc_pcxt->toc, fc_guc_len);
		SerializeGUCState(fc_guc_len, fc_gucspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_GUC, fc_gucspace);

		/* 序列化组合CID状态。 */
		fc_combocidspace = shm_toc_allocate(fc_pcxt->toc, fc_combocidlen);
		SerializeComboCIDState(fc_combocidlen, fc_combocidspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_COMBO_CID, fc_combocidspace);

		/*
		 * 如果事务隔离级别使用事务快照，则序列化事务快照。
		 */
		if (IsolationUsesXactSnapshot())
		{
			fc_tsnapspace = shm_toc_allocate(fc_pcxt->toc, fc_tsnaplen);
			SerializeSnapshot(fc_transaction_snapshot, fc_tsnapspace);
			shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT,
						   fc_tsnapspace);
		}

		/* 序列化活动快照。 */
		fc_asnapspace = shm_toc_allocate(fc_pcxt->toc, fc_asnaplen);
		SerializeSnapshot(fc_active_snapshot, fc_asnapspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, fc_asnapspace);

		/* 提供每个会话段的句柄。 */
		fc_session_dsm_handle_space = shm_toc_allocate(fc_pcxt->toc,
													sizeof(dsm_handle));
		*(dsm_handle *) fc_session_dsm_handle_space = fc_session_dsm_handle;
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_SESSION_DSM,
					   fc_session_dsm_handle_space);

		/* 序列化事务状态。 */
		fc_tstatespace = shm_toc_allocate(fc_pcxt->toc, fc_tstatelen);
		SerializeTransactionState(fc_tstatelen, fc_tstatespace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_TRANSACTION_STATE, fc_tstatespace);

		/* 序列化待处理同步。 */
		fc_pendingsyncsspace = shm_toc_allocate(fc_pcxt->toc, fc_pendingsyncslen);
		SerializePendingSyncs(fc_pendingsyncslen, fc_pendingsyncsspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_PENDING_SYNCS,
					   fc_pendingsyncsspace);

		/* 序列化重建索引状态。 */
		fc_reindexspace = shm_toc_allocate(fc_pcxt->toc, fc_reindexlen);
		SerializeReindexState(fc_reindexlen, fc_reindexspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_REINDEX_STATE, fc_reindexspace);

		/* 序列化关系映射器状态。 */
		fc_relmapperspace = shm_toc_allocate(fc_pcxt->toc, fc_relmapperlen);
		SerializeRelationMap(fc_relmapperlen, fc_relmapperspace);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_RELMAPPER_STATE,
					   fc_relmapperspace);

		/* 序列化未提交枚举状态。 */
		fc_uncommittedenumsspace = shm_toc_allocate(fc_pcxt->toc,
												 fc_uncommittedenumslen);
		SerializeUncommittedEnums(fc_uncommittedenumsspace, fc_uncommittedenumslen);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
					   fc_uncommittedenumsspace);

		/* 为工作进程信息分配空间。 */
		fc_pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * fc_pcxt->nworkers);

		/*
		 * 在动态共享内存中建立错误队列。
		 *
		 * 这些队列应仅用于传输ErrorResponse、NoticeResponse和NotifyResponse协议消息。元组数据应通过单独的（可能更大的？）队列传输。
		 */
		fc_error_queue_space =
			shm_toc_allocate(fc_pcxt->toc,
							 mul_size(PARALLEL_ERROR_QUEUE_SIZE,
									  fc_pcxt->nworkers));
		for (fc_i = 0; fc_i < fc_pcxt->nworkers; ++fc_i)
		{
			char	   *fc_start;
			shm_mq	   *fc_mq;

			fc_start = fc_error_queue_space + fc_i * PARALLEL_ERROR_QUEUE_SIZE;
			fc_mq = shm_mq_create(fc_start, PARALLEL_ERROR_QUEUE_SIZE);
			shm_mq_set_receiver(fc_mq, MyProc);
			fc_pcxt->worker[fc_i].error_mqh = shm_mq_attach(fc_mq, fc_pcxt->seg, NULL);
		}
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, fc_error_queue_space);

		/*
		 * 序列化入口点信息。通过进程传递函数指针是不安全的，因为在EXEC_BACKEND构建中，函数指针在每个进程中可能不同，因此我们始终传递库名和函数名。（我们使用库名“postgres”用于核心后端中的函数。）
		 */
		fc_lnamelen = strlen(fc_pcxt->library_name);
		fc_entrypointstate = shm_toc_allocate(fc_pcxt->toc, fc_lnamelen +
										   strlen(fc_pcxt->function_name) + 2);
		strcpy(fc_entrypointstate, fc_pcxt->library_name);
		strcpy(fc_entrypointstate + fc_lnamelen + 1, fc_pcxt->function_name);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_ENTRYPOINT, fc_entrypointstate);
	}

	/* 更新nworkers_to_launch，以防我们在上面更改了nworkers。 */
	fc_pcxt->nworkers_to_launch = fc_pcxt->nworkers;

	/* 恢复先前的内存上下文。 */
	MemoryContextSwitchTo(fc_oldcontext);
}

/*
 * 重新初始化动态共享内存段，以便我们可以再次为其启动工作进程。
 */
void ReinitializeParallelDSM(ParallelContext *fc_pcxt)
{
	FixedParallelState *fc_fps;

	/* 等待任何旧工作进程退出。 */
	if (fc_pcxt->nworkers_launched > 0)
	{
		WaitForParallelWorkersToFinish(fc_pcxt);
		fc_WaitForParallelWorkersToExit(fc_pcxt);
		fc_pcxt->nworkers_launched = 0;
		if (fc_pcxt->known_attached_workers)
		{
			pfree(fc_pcxt->known_attached_workers);
			fc_pcxt->known_attached_workers = NULL;
			fc_pcxt->nknown_attached_workers = 0;
		}
	}

	/* 将一些固定并行状态重置为干净状态。 */
	fc_fps = shm_toc_lookup(fc_pcxt->toc, PARALLEL_KEY_FIXED, false);
	fc_fps->last_xlog_end = 0;

	/* 重新创建错误队列（如果存在）。 */
	if (fc_pcxt->nworkers > 0)
	{
		char	   *fc_error_queue_space;
		int			fc_i;

		fc_error_queue_space =
			shm_toc_lookup(fc_pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, false);
		for (fc_i = 0; fc_i < fc_pcxt->nworkers; ++fc_i)
		{
			char	   *fc_start;
			shm_mq	   *fc_mq;

			fc_start = fc_error_queue_space + fc_i * PARALLEL_ERROR_QUEUE_SIZE;
			fc_mq = shm_mq_create(fc_start, PARALLEL_ERROR_QUEUE_SIZE);
			shm_mq_set_receiver(fc_mq, MyProc);
			fc_pcxt->worker[fc_i].error_mqh = shm_mq_attach(fc_mq, fc_pcxt->seg, NULL);
		}
	}
}

/*
 * 重新初始化并行上下文的并行工作者，以便我们可以启动不同数量的工作者。这在需要重用相同的DSM段但工作者数量可能因运行而异的情况下是必要的。
 */
void ReinitializeParallelWorkers(ParallelContext *fc_pcxt, int fc_nworkers_to_launch)
{
	/*
	 * 需要启动的工作者数量必须少于并行上下文初始化时的工作者数量。但是调用者可能不知道InitializeParallelDSM减少了nworkers，因此只需静默地调整请求。
	 */
	fc_pcxt->nworkers_to_launch = Min(fc_pcxt->nworkers, fc_nworkers_to_launch);
}

/*
 * 启动并行工作者。
 */
void LaunchParallelWorkers(ParallelContext *fc_pcxt)
{
	MemoryContext fc_oldcontext;
	BackgroundWorker fc_worker;
	int			fc_i;
	bool		fc_any_registrations_failed = false;

	/* 如果我们没有工作者则跳过此步骤。 */
	if (fc_pcxt->nworkers == 0 || fc_pcxt->nworkers_to_launch == 0)
		return;

	/* 我们需要成为一个锁组的领导者。 */
	BecomeLockGroupLeader();

	/* 如果我们有工作者，我们最好有一个DSM段。 */
	Assert(fc_pcxt->seg != NULL);

	/* 我们可能在一个短暂的内存上下文中运行。 */
	fc_oldcontext = MemoryContextSwitchTo(TopTransactionContext);

	/* 配置一个工作者。 */
	memset(&fc_worker, 0, sizeof(fc_worker));
	snprintf(fc_worker.bgw_name, BGW_MAXLEN, "parallel worker for PID %d",
			 MyProcPid);
	snprintf(fc_worker.bgw_type, BGW_MAXLEN, "parallel worker");
	fc_worker.bgw_flags =
		BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION
		| BGWORKER_CLASS_PARALLEL;
	fc_worker.bgw_start_time = BgWorkerStart_ConsistentState;
	fc_worker.bgw_restart_time = BGW_NEVER_RESTART;
	sprintf(fc_worker.bgw_library_name, "postgres");
	sprintf(fc_worker.bgw_function_name, "ParallelWorkerMain");
	fc_worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(fc_pcxt->seg));
	fc_worker.bgw_notify_pid = MyProcPid;

	/*
	 * 启动工作者。
	 *
	 * 调用者必须能够容忍最终得到的工作者数量少于预期，因此如果注册失败，这里没有必要抛出错误。因为在任何情况下，注册工作者并不保证它会成功启动和初始化。
	 */
	for (fc_i = 0; fc_i < fc_pcxt->nworkers_to_launch; ++fc_i)
	{
		memcpy(fc_worker.bgw_extra, &fc_i, sizeof(int));
		if (!fc_any_registrations_failed &&
			RegisterDynamicBackgroundWorker(&fc_worker,
											&fc_pcxt->worker[fc_i].bgwhandle))
		{
			shm_mq_set_handle(fc_pcxt->worker[fc_i].error_mqh,
							  fc_pcxt->worker[fc_i].bgwhandle);
			fc_pcxt->nworkers_launched++;
		}
		else
		{
			/*
			 * 如果我们无法注册工作者，那么我们已经达到了max_worker_processes限制，并且未来的注册可能也会失败，因此安排跳过它们。但我们仍然需要执行此代码以处理剩余的槽，以确保我们忘记为这些工作者预算的错误队列。否则，我们将等待它们启动，但它们永远不会启动。
			 */
			fc_any_registrations_failed = true;
			fc_pcxt->worker[fc_i].bgwhandle = NULL;
			shm_mq_detach(fc_pcxt->worker[fc_i].error_mqh);
			fc_pcxt->worker[fc_i].error_mqh = NULL;
		}
	}

	/*
	 * 现在nworkers_launched已经达到了最终值，我们可以初始化known_attached_workers。
	 */
	if (fc_pcxt->nworkers_launched > 0)
	{
		fc_pcxt->known_attached_workers =
			palloc0(sizeof(bool) * fc_pcxt->nworkers_launched);
		fc_pcxt->nknown_attached_workers = 0;
	}

	/* 恢复先前的内存上下文。 */
	MemoryContextSwitchTo(fc_oldcontext);
}

/*
 * 等待所有工作线程附加到它们的错误队列，如果
 * 有任何工作线程未能做到这一点，则抛出错误。
 *
 * 调用者可以假设如果此函数成功返回，则由
 * pcxt->nworkers_launched 给出的工作线程数量已初始化并
 * 附加到它们的错误队列。这些工作线程是否保证仍在运行，
 * 取决于调用者要求它们运行的代码；该函数不保证它们没有退出。
 * 但是，它确实保证任何已退出的工作线程必须是干净地退出，并
 * 且在成功执行它们的任务后退出。
 *
 * 如果此函数未被调用，则一些已启动的工作线程
 * 可能由于 fork() 失败而未能启动，或者在附加到
 * 错误队列之前在早期启动期间已退出，因此 nworkers_launched
 * 不能被视为完全可靠。它永远不会少于实际启动的工作线程数量，
 * 但可能会更多。任何未能启动的工作线程将在
 * WaitForParallelWorkersToFinish 中被发现，并在那时抛出错误，
 * 前提是最终调用到该函数。
 *
 * 一般来说，领导进程在调用此函数之前应该尽可能多地工作。
 * fork() 失败和其他早期启动失败是非常罕见的，而领导者在
 * 可以进行有用工作的情况下保持空闲是不理想的。然而，如果领导者
 * 需要等待所有工作线程或特定工作线程，它可能希望在执行此操作
 * 之前调用此函数。否则，它必须为未能启动的情况制定其他措施，
 * 以免永远等待。另一方面，领导者永远不等待可能尚未启动的
 * 工作线程，或者至少在 WaitForParallelWorkersToFinish() 之前
 * 从不这样做，也不需要调用此函数。
 */
void WaitForParallelWorkersToAttach(ParallelContext *fc_pcxt)
{
	int			fc_i;

	/* 如果没有启动的工作线程，则跳过此步骤。 */
	if (fc_pcxt->nworkers_launched == 0)
		return;

	for (;;)
	{
		/*
		 * 这将处理任何待处理的并行消息，并且它可能
		 * 还会抛出来自工作线程的传播错误。
		 */
		CHECK_FOR_INTERRUPTS();

		for (fc_i = 0; fc_i < fc_pcxt->nworkers_launched; ++fc_i)
		{
			BgwHandleStatus fc_status;
			shm_mq	   *fc_mq;
			int			fc_rc;
			pid_t		fc_pid;

			if (fc_pcxt->known_attached_workers[fc_i])
				continue;

			/*
			 * 如果 error_mqh 为 NULL，则表示工作线程已经
			 * 干净地退出。
			 */
			if (fc_pcxt->worker[fc_i].error_mqh == NULL)
			{
				fc_pcxt->known_attached_workers[fc_i] = true;
				++fc_pcxt->nknown_attached_workers;
				continue;
			}

			fc_status = GetBackgroundWorkerPid(fc_pcxt->worker[fc_i].bgwhandle, &fc_pid);
			if (fc_status == BGWH_STARTED)
			{
				/* 工作线程是否已附加到错误队列？ */
				fc_mq = shm_mq_get_queue(fc_pcxt->worker[fc_i].error_mqh);
				if (shm_mq_get_sender(fc_mq) != NULL)
				{
					/* 是的，因此它已知附加。 */
					fc_pcxt->known_attached_workers[fc_i] = true;
					++fc_pcxt->nknown_attached_workers;
				}
			}
			else if (fc_status == BGWH_STOPPED)
			{
				/*
				 * 如果工作线程未附加到错误队列而停止，
				 * 则抛出错误。
				 */
				fc_mq = shm_mq_get_queue(fc_pcxt->worker[fc_i].error_mqh);
				if (shm_mq_get_sender(fc_mq) == NULL)
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("parallel worker failed to initialize"),
							 errhint("More details may be available in the server log.")));

				fc_pcxt->known_attached_workers[fc_i] = true;
				++fc_pcxt->nknown_attached_workers;
			}
			else
			{
				/*
				 * 工作线程尚未启动，因此我们必须等待。主进程
				 * 会在工作线程的状态发生变化时通知我们。
				 * 我们的触发器也可能因其他原因被设置，但如果是这样，
				 * 我们将再次等待同一工作线程。
				 */
				fc_rc = WaitLatch(MyLatch,
							   WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
							   -1, WAIT_EVENT_BGWORKER_STARTUP);

				if (fc_rc & WL_LATCH_SET)
					ResetLatch(MyLatch);
			}
		}

		/* 如果所有工作线程已知已启动，则我们完成了。 */
		if (fc_pcxt->nknown_attached_workers >= fc_pcxt->nworkers_launched)
		{
			Assert(fc_pcxt->nknown_attached_workers == fc_pcxt->nworkers_launched);
			break;
		}
	}
}

/*
 * 等待所有工作线程完成计算。
 *
 * 即使并行操作似乎已经成功完成，但在此之后调用此函数仍然很重要。我们必须确保不会遗漏工作线程在并行操作期间可能抛出的任何错误，或者在关闭时可能会抛出的任何错误。
 *
 * 此外，我们希望根据工作线程的反馈更新我们对 XactLastRecEnd 的概念。
 */
void WaitForParallelWorkersToFinish(ParallelContext *fc_pcxt)
{
	for (;;)
	{
		bool		fc_anyone_alive = false;
		int			fc_nfinished = 0;
		int			fc_i;

		/*
		 * 这将处理任何待处理的并行消息，这可能会改变后续循环的结果。它还可能抛出来自工作线程的传播错误。
		 */
		CHECK_FOR_INTERRUPTS();

		for (fc_i = 0; fc_i < fc_pcxt->nworkers_launched; ++fc_i)
		{
			/*
			 * 如果 error_mqh 为 NULL，则工作线程已经正常退出。如果我们通过 error_mqh 从工作线程接收到消息，我们知道它已正常启动，因此我们可以确定在它退出时会收到通知。
			 */
			if (fc_pcxt->worker[fc_i].error_mqh == NULL)
				++fc_nfinished;
			else if (fc_pcxt->known_attached_workers[fc_i])
			{
				fc_anyone_alive = true;
				break;
			}
		}

		if (!fc_anyone_alive)
		{
			/* 如果所有工作线程都已知已完成，我们就完成了。 */
			if (fc_nfinished >= fc_pcxt->nworkers_launched)
			{
				Assert(fc_nfinished == fc_pcxt->nworkers_launched);
				break;
			}

			/*
			 * 我们没有检测到任何活着的工作线程，但并不是所有工作线程都已知已正常退出。可能并不是所有工作线程都已启动，或者它们中的某些线程未能启动或异常终止。
			 */
			for (fc_i = 0; fc_i < fc_pcxt->nworkers_launched; ++fc_i)
			{
				pid_t		fc_pid;
				shm_mq	   *fc_mq;

				/*
				 * 如果工作线程是 BGWH_NOT_YET_STARTED 或 BGWH_STARTED，我们应该继续等待。如果它是 BGWH_STOPPED，则需要进一步调查。
				 */
				if (fc_pcxt->worker[fc_i].error_mqh == NULL ||
					fc_pcxt->worker[fc_i].bgwhandle == NULL ||
					GetBackgroundWorkerPid(fc_pcxt->worker[fc_i].bgwhandle,
										   &fc_pid) != BGWH_STOPPED)
					continue;

				/*
				 * 检查工作线程是否在未附加到错误队列的情况下停止。如果是这样，主进程无法派生工作线程或其未正确初始化而退出。我们必须抛出错误，因为调用者可能期望工作线程在退出前执行一些工作。
				 */
				fc_mq = shm_mq_get_queue(fc_pcxt->worker[fc_i].error_mqh);
				if (shm_mq_get_sender(fc_mq) == NULL)
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("parallel worker failed to initialize"),
							 errhint("More details may be available in the server log.")));

				/*
				 * 工作线程已停止，但已附加到错误队列。除非某处存在错误，否则这种情况只会发生在工作线程在本函数顶部的 CHECK_FOR_INTERRUPTS() 之后写入消息并终止之前。在这种情况下，或信号量也应该已被设置，在下一次循环中会发生正确的事情。
				 */
			}
		}

		(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, -1,
						 WAIT_EVENT_PARALLEL_FINISH);
		ResetLatch(MyLatch);
	}

	if (fc_pcxt->toc != NULL)
	{
		FixedParallelState *fc_fps;

		fc_fps = shm_toc_lookup(fc_pcxt->toc, PARALLEL_KEY_FIXED, false);
		if (fc_fps->last_xlog_end > XactLastRecEnd)
			XactLastRecEnd = fc_fps->last_xlog_end;
	}
}

/*
 * 等待所有工作线程退出。
 *
 * 此函数确保工作线程已完全关闭。WaitForParallelWorkersToFinish 和此函数之间的区别在于，前者仅确保工作后端发送的最后一条消息已被领导后端接收，而此函数确保完全关闭。
 */
static void fc_WaitForParallelWorkersToExit(ParallelContext *fc_pcxt)
{
	int			fc_i;

	/* 等待工作线程实际死亡。 */
	for (fc_i = 0; fc_i < fc_pcxt->nworkers_launched; ++fc_i)
	{
		BgwHandleStatus fc_status;

		if (fc_pcxt->worker == NULL || fc_pcxt->worker[fc_i].bgwhandle == NULL)
			continue;

		fc_status = WaitForBackgroundWorkerShutdown(fc_pcxt->worker[fc_i].bgwhandle);

		/*
		 * 如果主进程终止，那么我们就无法安全清理——我们无法知道我们的工作线程何时真正死亡。这并不需要 PANIC，因为它们最终都会中止，但我们无法安全地继续此会话。
		 */
		if (fc_status == BGWH_POSTMASTER_DIED)
			ereport(FATAL,
					(errcode(ERRCODE_ADMIN_SHUTDOWN),
					 errmsg("postmaster exited during a parallel transaction")));

		/* 释放内存。 */
		pfree(fc_pcxt->worker[fc_i].bgwhandle);
		fc_pcxt->worker[fc_i].bgwhandle = NULL;
	}
}

/* 
 * 销毁并行上下文。
 *
 * 如果期望干净退出，您应该先使用 WaitForParallelWorkersToFinish() 
 * 再调用此函数。当调用此函数时，任何剩余的工作进程都将被强制终止；动态共享内存段 
 * 被解除映射；然后我们将等待（不可中断地）工作进程退出。
 */
void DestroyParallelContext(ParallelContext *fc_pcxt)
{
	int			fc_i;

	/*
	 * 在这里要小心操作顺序！我们在做其他任何事情之前先从列表中移除 
	 * 并行上下文；否则，如果在后续步骤中发生错误，我们可能会尝试从 
	 * AtEOXact_Parallel 或 AtEOSubXact_Parallel 再次终止它。
	 */
	dlist_delete(&fc_pcxt->node);

	/* 逐个终止每个工作进程，并忘记它们的错误队列。 */
	if (fc_pcxt->worker != NULL)
	{
		for (fc_i = 0; fc_i < fc_pcxt->nworkers_launched; ++fc_i)
		{
			if (fc_pcxt->worker[fc_i].error_mqh != NULL)
			{
				TerminateBackgroundWorker(fc_pcxt->worker[fc_i].bgwhandle);

				shm_mq_detach(fc_pcxt->worker[fc_i].error_mqh);
				fc_pcxt->worker[fc_i].error_mqh = NULL;
			}
		}
	}

	/*
	 * 如果我们已经分配了共享内存段，解除它的连接。这将 
	 * 隐式地解除错误队列和存储在那里的任何其他共享内存队列的连接。
	 */
	if (fc_pcxt->seg != NULL)
	{
		dsm_detach(fc_pcxt->seg);
		fc_pcxt->seg = NULL;
	}

	/*
	 * 如果该并行上下文实际上是在后端私有内存中而不是共享内存中， 
	 * 则释放该内存。
	 */
	if (fc_pcxt->private_memory != NULL)
	{
		pfree(fc_pcxt->private_memory);
		fc_pcxt->private_memory = NULL;
	}

	/*
	 * 在所有工作进程退出之前，我们无法完成事务提交或中止。这意味着， 
	 * 尤其是，我们在这个阶段无法响应中断。
	 */
	HOLD_INTERRUPTS();
	fc_WaitForParallelWorkersToExit(fc_pcxt);
	RESUME_INTERRUPTS();

	/* 释放工作进程数组本身。 */
	if (fc_pcxt->worker != NULL)
	{
		pfree(fc_pcxt->worker);
		fc_pcxt->worker = NULL;
	}

	/* 释放内存。 */
	pfree(fc_pcxt->library_name);
	pfree(fc_pcxt->function_name);
	pfree(fc_pcxt);
}

/*
 * 当前是否有任何并行上下文处于活动状态？
 */
bool ParallelContextActive(void)
{
	return !dlist_is_empty(&pcxt_list);
}

/*
 * 处理接收到的指示并行工作进程消息的中断。
 *
 * 注意：这是在信号处理程序内调用的！我们能做的只是设置 
 * 一个标志，使得下一个 CHECK_FOR_INTERRUPTS() 调用 
 * HandleParallelMessages()。
 */
void HandleParallelMessageInterrupt(void)
{
	InterruptPending = true;
	ParallelMessagePending = true;
	SetLatch(MyLatch);
}

/*
 * 处理从并行工作进程接收到的任何排队协议消息。
 */
void HandleParallelMessages(void)
{
	dlist_iter	fc_iter;
	MemoryContext fc_oldcontext;

	static MemoryContext fc_hpm_context = NULL;

	/*
	 * 这是从 ProcessInterrupts() 调用的，由于它调用的一些 
	 * 函数包含 CHECK_FOR_INTERRUPTS()，因此如果在此运行时接收到更多信号， 
	 * 可能会出现递归调用。尚不清楚递归进入是否安全，即使安全似乎也没有用处， 
	 * 所以在完成之前我们暂时阻止中断。
	 */
	HOLD_INTERRUPTS();

	/*
	 * 此外，CurrentMemoryContext 可能几乎指向任何地方。我们 
	 * 不想冒险将数据泄漏到长寿命上下文中，因此让我们 
	 * 在这里做我们的工作，在一个可以在每次使用时重置的私有上下文中。
	 */
	if (fc_hpm_context == NULL)	/* 第一次通过？ */
		fc_hpm_context = AllocSetContextCreate(TopMemoryContext,
											"HandleParallelMessages",
											ALLOCSET_DEFAULT_SIZES);
	else
		MemoryContextReset(fc_hpm_context);

	fc_oldcontext = MemoryContextSwitchTo(fc_hpm_context);

	/* 可以处理消息。重置表示还有更多需处理的标志。 */
	ParallelMessagePending = false;

	dlist_foreach(fc_iter, &pcxt_list)
	{
		ParallelContext *fc_pcxt;
		int			fc_i;

		fc_pcxt = dlist_container(ParallelContext, node, fc_iter.cur);
		if (fc_pcxt->worker == NULL)
			continue;

		for (fc_i = 0; fc_i < fc_pcxt->nworkers_launched; ++fc_i)
		{
			/*
			 * 从每个工作进程读取尽可能多的消息，但在以下情况下停止：
			 * （1）工作进程的错误队列消失，这可能发生 
			 * 如果我们收到来自工作进程的终止消息；或（2）不能 
			 * 再从工作进程读取更多消息而不阻塞。
			 */
			while (fc_pcxt->worker[fc_i].error_mqh != NULL)
			{
				shm_mq_result fc_res;
				Size		fc_nbytes;
				void	   *fc_data;

				fc_res = shm_mq_receive(fc_pcxt->worker[fc_i].error_mqh, &fc_nbytes,
									 &fc_data, true);
				if (fc_res == SHM_MQ_WOULD_BLOCK)
					break;
				else if (fc_res == SHM_MQ_SUCCESS)
				{
					StringInfoData fc_msg;

					initStringInfo(&fc_msg);
					appendBinaryStringInfo(&fc_msg, fc_data, fc_nbytes);
					fc_HandleParallelMessage(fc_pcxt, fc_i, &fc_msg);
					pfree(fc_msg.data);
				}
				else
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("lost connection to parallel worker")));
			}
		}
	}

	MemoryContextSwitchTo(fc_oldcontext);

	/* 走的时候不妨清除上下文 */
	MemoryContextReset(fc_hpm_context);

	RESUME_INTERRUPTS();
}

/*
 * 处理从单个并行工作进程接收到的单个协议消息。
 */
static void fc_HandleParallelMessage(ParallelContext *fc_pcxt, int fc_i, StringInfo fc_msg)
{
	char		fc_msgtype;

	if (fc_pcxt->known_attached_workers != NULL &&
		!fc_pcxt->known_attached_workers[fc_i])
	{
		fc_pcxt->known_attached_workers[fc_i] = true;
		fc_pcxt->nknown_attached_workers++;
	}

	fc_msgtype = pq_getmsgbyte(fc_msg);

	switch (fc_msgtype)
	{
		case 'K':				/* BackendKeyData */
			{
				int32		fc_pid = pq_getmsgint(fc_msg, 4);

				(void) pq_getmsgint(fc_msg, 4);	/* 丢弃取消键 */
				(void) pq_getmsgend(fc_msg);
				fc_pcxt->worker[fc_i].pid = fc_pid;
				break;
			}

		case 'E':				/* ErrorResponse */
		case 'N':				/* NoticeResponse */
			{
				ErrorData	fc_edata;
				ErrorContextCallback *fc_save_error_context_stack;

				/* 解析 ErrorResponse 或 NoticeResponse。 */
				pq_parse_errornotice(fc_msg, &fc_edata);

				/* 工人的死亡不足以成为自杀的合理理由。 */
				fc_edata.elevel = Min(fc_edata.elevel, ERROR);

				/*
				 * 如果需要，可以添加一行上下文以显示这是
				 * 从并行工作线程传播过来的消息。否则，有时
				 * 理解实际发生的事情会有些困惑。（我们在FORCE_PARALLEL_REGRESS模式下不这样做，
				 * 因为这会导致测试结果的不稳定，具体取决于
				 * 是否实际使用了并行工作线程。）
				 */
				if (force_parallel_mode != FORCE_PARALLEL_REGRESS)
				{
					if (fc_edata.context)
						fc_edata.context = psprintf("%s\n%s", fc_edata.context,
												 _("parallel worker"));
					else
						fc_edata.context = pstrdup(_("parallel worker"));
				}

				/*
				 * 上下文之外的内容应该使用在创建ParallelContext时
				 * 生效的错误上下文回调，而不是当前的回调。
				 */
				fc_save_error_context_stack = error_context_stack;
				error_context_stack = fc_pcxt->error_context_stack;

				/* 重新抛出错误或打印通知。 */
				ThrowErrorData(&fc_edata);

				/* 不是错误，所以恢复之前的上下文栈。 */
				error_context_stack = fc_save_error_context_stack;

				break;
			}

		case 'A':				/* NotifyResponse */
			{
				/* 传播NotifyResponse。 */
				int32		fc_pid;
				const char *fc_channel;
				const char *fc_payload;

				fc_pid = pq_getmsgint(fc_msg, 4);
				fc_channel = pq_getmsgrawstring(fc_msg);
				fc_payload = pq_getmsgrawstring(fc_msg);
				pq_endmessage(fc_msg);

				NotifyMyFrontEnd(fc_channel, fc_payload, fc_pid);

				break;
			}

		case 'X':				/* 终止，表示正常退出 */
			{
				shm_mq_detach(fc_pcxt->worker[fc_i].error_mqh);
				fc_pcxt->worker[fc_i].error_mqh = NULL;
				break;
			}

		default:
			{
				elog(ERROR, "unrecognized message type received from parallel worker: %c (message length %d bytes)",
					 fc_msgtype, fc_msg->len);
			}
	}
}

/*
 * 并行上下文的子事务结束时的清理。
 *
 * 目前，禁止在并行模式下进入或离开子事务，因此我们可以直接清除所有内容。但
 * 我们可能希望在未来放宽这一限制，因此这段代码
 * 考虑到pcxt_list中可能有多个子事务ID。
 */
void AtEOSubXact_Parallel(bool fc_isCommit, SubTransactionId fc_mySubId)
{
	while (!dlist_is_empty(&pcxt_list))
	{
		ParallelContext *fc_pcxt;

		fc_pcxt = dlist_head_element(ParallelContext, node, &pcxt_list);
		if (fc_pcxt->subid != fc_mySubId)
			break;
		if (fc_isCommit)
			elog(WARNING, "leaked parallel context");
		DestroyParallelContext(fc_pcxt);
	}
}

/*
 * 并行上下文的事务结束时的清理。
 */
void AtEOXact_Parallel(bool fc_isCommit)
{
	while (!dlist_is_empty(&pcxt_list))
	{
		ParallelContext *fc_pcxt;

		fc_pcxt = dlist_head_element(ParallelContext, node, &pcxt_list);
		if (fc_isCommit)
			elog(WARNING, "leaked parallel context");
		DestroyParallelContext(fc_pcxt);
	}
}

/*
 * 并行工作线程的主要入口点。
 */
void ParallelWorkerMain(Datum fc_main_arg)
{
	dsm_segment *fc_seg;
	shm_toc    *fc_toc;
	FixedParallelState *fc_fps;
	char	   *fc_error_queue_space;
	shm_mq	   *fc_mq;
	shm_mq_handle *fc_mqh;
	char	   *fc_libraryspace;
	char	   *fc_entrypointstate;
	char	   *fc_library_name;
	char	   *fc_function_name;
	parallel_worker_main_type fc_entrypt;
	char	   *fc_gucspace;
	char	   *fc_combocidspace;
	char	   *fc_tsnapspace;
	char	   *fc_asnapspace;
	char	   *fc_tstatespace;
	char	   *fc_pendingsyncsspace;
	char	   *fc_reindexspace;
	char	   *fc_relmapperspace;
	char	   *fc_uncommittedenumsspace;
	StringInfoData fc_msgbuf;
	char	   *fc_session_dsm_handle_space;
	Snapshot	fc_tsnapshot;
	Snapshot	fc_asnapshot;

	/* 设置标记以表示我们正在初始化一个并行工作线程。 */
	InitializingParallelWorker = true;

	/* 建立信号处理器。 */
	pqsignal(SIGTERM, die);
	BackgroundWorkerUnblockSignals();

	/* 确定并设置我们的并行工作线程编号。 */
	Assert(ParallelWorkerNumber == -1);
	memcpy(&ParallelWorkerNumber, MyBgworkerEntry->bgw_extra, sizeof(int));

	/* 为了整洁，设置一个内存上下文进行工作。 */
	CurrentMemoryContext = AllocSetContextCreate(TopMemoryContext,
												 "Parallel worker",
												 ALLOCSET_DEFAULT_SIZES);

	/*
	 * 附加到并行查询的动态共享内存段，并
	 * 找到它的目录。
	 *
	 * 注意：在这一点上，我们尚未在该进程中创建任何ResourceOwner。
	 * 这将导致我们的DSM映射在进程退出之前保持有效，这很好。
	 * 如果有ResourceOwner，它会获得映射的所有权，
	 * 但我们并不需要这样做。
	 */
	fc_seg = dsm_attach(DatumGetUInt32(fc_main_arg));
	if (fc_seg == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("could not map dynamic shared memory segment")));
	fc_toc = shm_toc_attach(PARALLEL_MAGIC, dsm_segment_address(fc_seg));
	if (fc_toc == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("invalid magic number in dynamic shared memory segment")));

	/* 查找固定的并行状态。 */
	fc_fps = shm_toc_lookup(fc_toc, PARALLEL_KEY_FIXED, false);
	MyFixedParallelState = fc_fps;

	/* 安排在退出时向领导者发送信号。 */
	ParallelLeaderPid = fc_fps->parallel_leader_pid;
	ParallelLeaderBackendId = fc_fps->parallel_leader_backend_id;
	before_shmem_exit(fc_ParallelWorkerShutdown, PointerGetDatum(fc_seg));

	/*
	 * 现在我们可以找到并附加到为我们提供的错误队列。
	 * 这是好的，因为在我们做到这一点之前，这里发生的任何错误都不会
	 * 报告回请求启动该工作线程的进程。
	 */
	fc_error_queue_space = shm_toc_lookup(fc_toc, PARALLEL_KEY_ERROR_QUEUE, false);
	fc_mq = (shm_mq *) (fc_error_queue_space +
					 ParallelWorkerNumber * PARALLEL_ERROR_QUEUE_SIZE);
	shm_mq_set_sender(fc_mq, MyProc);
	fc_mqh = shm_mq_attach(fc_mq, fc_seg, NULL);
	pq_redirect_to_shm_mq(fc_seg, fc_mqh);
	pq_set_parallel_leader(fc_fps->parallel_leader_pid,
						   fc_fps->parallel_leader_backend_id);

	/*
	 * 向发起并行的进程发送BackendKeyData消息，
	 * 以便在接收我们发送的任何其他消息之前能够访问我们的PID。
	 * 我们的取消键也会被发送，因为这是协议消息的定义方式，
	 * 但在这种情况下它实际上不会被用于任何事情。
	 */
	pq_beginmessage(&fc_msgbuf, 'K');
	pq_sendint32(&fc_msgbuf, (int32) MyProcPid);
	pq_sendint32(&fc_msgbuf, (int32) MyCancelKey);
	pq_endmessage(&fc_msgbuf);

	/*
	 * 好极了！主要初始化已完成。现在，我们需要设置
	 * 我们的后端本地状态以匹配原始后端。
	 */

	/*
	 * 加入锁定组。在任何可能尝试获取重型锁的操作之前，必须执行此步骤，因为到此为止获取的任何重型锁可能会直接阻塞并行组的领导者或一些过程中等待与并行组领导者冲突的锁，导致未检测到的死锁。（如果我们无法加入锁定组，领导者已经离开，因此只需安静地退出。）
	 */
	if (!BecomeLockGroupMember(fc_fps->parallel_leader_pgproc,
							   fc_fps->parallel_leader_pid))
		return;

	/*
	 * 恢复事务和语句开始时间戳。必须在启动任何事务的操作之前进行此步骤，否则在xact.c中的断言将被触发。
	 */
	SetParallelStartTimestamps(fc_fps->xact_ts, fc_fps->stmt_ts);

	/*
	 * 确定要调用的入口点。从理论上讲，这可能导致加载额外的库，尽管大多数情况下入口点在核心后端或我们刚加载的库中。
	 */
	fc_entrypointstate = shm_toc_lookup(fc_toc, PARALLEL_KEY_ENTRYPOINT, false);
	fc_library_name = fc_entrypointstate;
	fc_function_name = fc_entrypointstate + strlen(fc_library_name) + 1;

	fc_entrypt = fc_LookupParallelWorkerFunction(fc_library_name, fc_function_name);

	/*
	 * 恢复当前会话的授权和角色 ID。这里没有验证，我们只是盲目采用领导者的状态。请注意，这必须在InitPostgres之前发生，因为InitializeSessionUserId不会设置这些变量。
	 */
	SetAuthenticatedUserId(fc_fps->authenticated_user_id,
						   fc_fps->authenticated_user_is_superuser);
	SetSessionAuthorization(fc_fps->session_user_id,
							fc_fps->session_user_is_superuser);
	SetCurrentRoleId(fc_fps->outer_user_id, fc_fps->role_is_superuser);

	/*
	 * 恢复数据库连接。我们跳过连接授权检查，推测（a）领导者在启动时检查了这些内容，以及（b）我们不希望并行模式导致这些故障，因为那样会使并行查询计划对应用程序不透明。
	 */
	BackgroundWorkerInitializeConnectionByOid(fc_fps->database_id,
											  fc_fps->authenticated_user_id,
											  BGWORKER_BYPASS_ALLOWCONN);

	/*
	 * 将客户端编码设置为数据库编码，因为这正是领导者所期望的。
	 */
	SetClientEncoding(GetDatabaseEncoding());

	/*
	 * 加载原始后端加载的库。我们希望在恢复GUC之前进行此操作，因为这些库可能定义自定义变量。
	 */
	fc_libraryspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_LIBRARY, false);
	StartTransactionCommand();
	RestoreLibraryState(fc_libraryspace);

	/* 从启动后端恢复GUC值。 */
	fc_gucspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_GUC, false);
	RestoreGUCState(fc_gucspace);
	CommitTransactionCommand();

	/* 启动适合并行工作者的事务状态。 */
	fc_tstatespace = shm_toc_lookup(fc_toc, PARALLEL_KEY_TRANSACTION_STATE, false);
	StartParallelWorkerTransaction(fc_tstatespace);

	/* 恢复组合CID状态。 */
	fc_combocidspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_COMBO_CID, false);
	RestoreComboCIDState(fc_combocidspace);

	/* 附加到每个会话的DSM段及其包含的对象。 */
	fc_session_dsm_handle_space =
		shm_toc_lookup(fc_toc, PARALLEL_KEY_SESSION_DSM, false);
	AttachSession(*(dsm_handle *) fc_session_dsm_handle_space);

	/*
	 * 如果事务隔离级别为REPEATABLE READ或SERIALIZABLE，领导者已序列化事务快照，我们必须恢复它。在较低的隔离级别下，没有事务生命周期的快照，但我们需要将TransactionXmin设置为不大于此工作者将使用的每个快照的xmin的值。实现这一点最简单的方法是将活动快照安装为事务快照。在此并行工作者中运行的代码可能会通过GetTransactionSnapshot()或GetLatestSnapshot()获取新的快照，但不应该有任何方式获取比活动快照更旧的快照。
	 */
	fc_asnapspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, false);
	fc_tsnapspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, true);
	fc_asnapshot = RestoreSnapshot(fc_asnapspace);
	fc_tsnapshot = fc_tsnapspace ? RestoreSnapshot(fc_tsnapspace) : fc_asnapshot;
	RestoreTransactionSnapshot(fc_tsnapshot,
							   fc_fps->parallel_leader_pgproc);
	PushActiveSnapshot(fc_asnapshot);

	/*
	 * 我们已更改可以看到的元组，因此必须使系统缓存失效。
	 */
	InvalidateSystemCaches();

	/*
	 * 恢复当前用户 ID 和安全上下文。这里不进行验证，
	 * 我们只是盲目采用领导者的状态。在恢复 GUCs 之后才能执行此操作，
	 * 否则我们会收到关于恢复 session_authorization 和角色的投诉。
	 * （实际上，我们假设所有恢复的值都是可以设置的，即使我们现在
	 * 处于受限上下文中。）
	 */
	SetUserIdAndSecContext(fc_fps->current_user_id, fc_fps->sec_context);

	/* 恢复临时命名空间状态以确保搜索路径与领导者匹配。 */
	SetTempNamespaceState(fc_fps->temp_namespace_id,
						  fc_fps->temp_toast_namespace_id);

	/* 恢复待处理的同步。 */
	fc_pendingsyncsspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_PENDING_SYNCS,
									   false);
	RestorePendingSyncs(fc_pendingsyncsspace);

	/* 恢复重建索引状态。 */
	fc_reindexspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_REINDEX_STATE, false);
	RestoreReindexState(fc_reindexspace);

	/* 恢复 relmapper 状态。 */
	fc_relmapperspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_RELMAPPER_STATE, false);
	RestoreRelationMap(fc_relmapperspace);

	/* 恢复未提交的枚举。 */
	fc_uncommittedenumsspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
										   false);
	RestoreUncommittedEnums(fc_uncommittedenumsspace);

	/* 如果是 SERIALIZABLE，则附加到领导者的可序列化事务。 */
	AttachSerializableXact(fc_fps->serializable_xact_handle);

	/*
	 * 我们现在已经初始化了所有状态；此后的情况不应改变。
	 */
	InitializingParallelWorker = false;
	EnterParallelMode();

	/*
	 * 是时候做真正的工作了：调用调用者提供的代码。
	 */
	fc_entrypt(fc_seg, fc_toc);

	/* 必须退出并行模式以弹出活动快照。 */
	ExitParallelMode();

	/* 必须弹出活动快照以便 snapmgr.c 不会抱怨。 */
	PopActiveSnapshot();

	/* 关闭并行工作者事务。 */
	EndParallelWorkerTransaction();

	/* 与每个会话的 DSM 段分离。 */
	DetachSession();

	/* 报告成功。 */
	pq_putmessage('X', NULL, 0);
}

/*
 * 如果我们写入的最后一个 WAL 记录的结束位置大于
 * 已经存储的值，更新共享内存。
 */
void ParallelWorkerReportLastRecEnd(XLogRecPtr fc_last_xlog_end)
{
	FixedParallelState *fc_fps = MyFixedParallelState;

	Assert(fc_fps != NULL);
	SpinLockAcquire(&fc_fps->mutex);
	if (fc_fps->last_xlog_end < fc_last_xlog_end)
		fc_fps->last_xlog_end = fc_last_xlog_end;
	SpinLockRelease(&fc_fps->mutex);
}

/*
 * 确保领导者再尝试从我们的错误队列读取一次。
 * 这可以防止我们在未发送 ErrorResponse 给领导者的情况下
 * 不干净地退出，例如因为某些代码直接调用了 proc_exit。
 *
 * 同时显式从 dsm 段分离，以便使用 on_dsm_detach() 的子系统
 * 有机会在 stats 子系统作为 before_shmem_exit() 钩子的一部分关闭之前
 * 发送统计信息。
 *
 * 有人可能认为这可以通过谨慎地安排附加到 dsm 段的顺序来解决，
 * 以便 pgstats 段在并行查询段之后被分离。然而，这并不奏效，
 * 因为 stats hash 可能需要增长，这可能会导致分配新段， 
 * 然后将其从较早的段分离。
 */
static void fc_ParallelWorkerShutdown(int fc_code, Datum fc_arg)
{
	SendProcSignal(ParallelLeaderPid,
				   PROCSIG_PARALLEL_MESSAGE,
				   ParallelLeaderBackendId);

	dsm_detach((dsm_segment *) DatumGetPointer(fc_arg));
}

/*
 * 查找（并可能加载）并行工作线程入口点函数。
 *
 * 对于包含在核心代码中的函数，我们使用库名称“postgres”
 * 并查询InternalParallelWorkers数组。外部函数则使用load_external_function()
 * 进行查找，并在必要时加载。
 *
 * 这样做的目的是为了跨进程边界传递函数名称作为字符串。
 * 我们不能传递实际的函数地址，因其可能已在不同进程中
 * 以不同地址加载。对于可加载库中的函数，这显然是一个危险，
 * 但即使在使用EXEC_BACKEND的平台上的核心代码中的函数
 * 也可能发生这种情况（例如，Windows）。
 *
 * 在某些时候，可能值得摆脱InternalParallelWorkers[]
 * 而选择对核心函数也应用load_external_function();
 * 但这会引发不值得现在解决的可移植性问题。
 */
static parallel_worker_main_type
fc_LookupParallelWorkerFunction(const char *fc_libraryname, const char *fc_funcname)
{
	/*
	 * 如果函数是从postgres自身加载，搜索
	 * InternalParallelWorkers数组。
	 */
	if (strcmp(fc_libraryname, "postgres") == 0)
	{
		int			fc_i;

		for (fc_i = 0; fc_i < lengthof(InternalParallelWorkers); fc_i++)
		{
			if (strcmp(InternalParallelWorkers[fc_i].fn_name, fc_funcname) == 0)
				return InternalParallelWorkers[fc_i].fn_addr;
		}

		/* 我们只能通过编程错误达到这一点。 */
		elog(ERROR, "internal function \"%s\" not found", fc_funcname);
	}

	/* 否则从外部库加载。 */
	return (parallel_worker_main_type)
		load_external_function(fc_libraryname, fc_funcname, true, NULL);
}
